package org.mq.mymq.broker.event.listener;

import com.alibaba.fastjson.JSON;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.event.model.CreateTopicEvent;
import org.mq.mymq.broker.model.CommitLogModel;
import org.mq.mymq.broker.model.MyMqTopicModel;
import org.mq.mymq.broker.model.QueueModel;
import org.mq.mymq.broker.utils.LogFileNameUtil;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.constants.BrokerConstants;
import org.mq.mymq.common.dto.CreateTopicReqDTO;
import org.mq.mymq.common.enums.BrokerClusterModeEnum;
import org.mq.mymq.common.enums.BrokerEventCode;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.common.utils.AssertUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName CreateTopicListener
 * @Description 创建Topic监听器 如果是master节点，创建topic后转发创建请求到slave
 * @Author jiarong_ye
 * @Date 2025/7/9 16:45
 * @Version 1.0
 */
public class CreateTopicListener implements Listener<CreateTopicEvent> {

    private static final Logger log = LoggerFactory.getLogger(CreateTopicListener.class);

    @Override
    public void onReceive(CreateTopicEvent event) throws Exception {
        CreateTopicReqDTO createTopicReqDTO = event.getCreateTopicReqDTO();
        AssertUtil.isTrue(createTopicReqDTO.getQueueSize() > 0 && createTopicReqDTO.getQueueSize() < 100, "queueSize参数异常");
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(createTopicReqDTO.getTopic());
        AssertUtil.isTrue(myMqTopicModel == null, "topic已经存在");
        createTopicFile(createTopicReqDTO);
        addTopicInCommonCache(createTopicReqDTO);
        loadFileInMMap(createTopicReqDTO);
        log.info("topic:{} is created! queueSize is {}", createTopicReqDTO.getTopic(), createTopicReqDTO.getQueueSize());
        event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CREATED_TOPIC_SUCCESS.getCode(), event.getMsgId().getBytes()));
        // 如果是主从同步架构且当前节点为master节点，触发数据同步，将数据发送到slave节点
        if (BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(CommonCache.getGlobalProperties().getBrokerClusterMode()) 
                && "master".equals(CommonCache.getGlobalProperties().getBrokerClusterRole())) {
            CommonCache.getSlaveChannelMap().values().forEach(channel -> {
                channel.writeAndFlush(new TcpMsg(BrokerEventCode.CREATE_TOPIC.getCode(), JSON.toJSONBytes(createTopicReqDTO)));
            });
        }
    }

    /**
     * 创建topic对应的文件
     *
     * @param createTopicReqDTO
     * @throws IOException
     */
    public static void createTopicFile(CreateTopicReqDTO createTopicReqDTO) throws IOException {
        String baseCommitLogDirPath = LogFileNameUtil.buildCommitLogBasePath(createTopicReqDTO.getTopic());
        File commitLogDir = new File(baseCommitLogDirPath);
        commitLogDir.mkdir();
        File commitLogFile = new File(baseCommitLogDirPath + BrokerConstants.SPLIT + LogFileNameUtil.buildFirstCommitLogName());
        commitLogFile.createNewFile();
        String baseConsumeQueueDirPath = LogFileNameUtil.buildConsumeQueueBasePath(createTopicReqDTO.getTopic());
        File consumeQueueFile = new File(baseConsumeQueueDirPath);
        consumeQueueFile.mkdir();
        for (Integer i = 0; i < createTopicReqDTO.getQueueSize(); i++) {
            String basePath = baseConsumeQueueDirPath + BrokerConstants.SPLIT + i;
            new File(basePath).mkdir();
            new File(basePath + BrokerConstants.SPLIT + LogFileNameUtil.buildFirstConsumeQueueName()).createNewFile();
        }
    }

    /**
     * 加载创建的commitLog和consumeQueue文件到mmap中
     *
     * @param createTopicReqDTO
     * @throws IOException
     */
    public static void loadFileInMMap(CreateTopicReqDTO createTopicReqDTO) throws IOException {
        String topic = createTopicReqDTO.getTopic();
        CommonCache.getCommitLogAppendHandler().prepareMMapLoading(topic);
        CommonCache.getConsumeQueueAppendHandler().prepareConsumeQueue(topic);
    }

    /**
     * 添加Topic信息到缓存中
     *
     * @param createTopicReqDTO
     */
    public static void addTopicInCommonCache(CreateTopicReqDTO createTopicReqDTO) {
        MyMqTopicModel myMqTopicModel = new MyMqTopicModel();
        myMqTopicModel.setTopic(createTopicReqDTO.getTopic());
        long currentTime = System.currentTimeMillis();
        myMqTopicModel.setCreateAt(currentTime);
        myMqTopicModel.setUpdateAt(currentTime);
        CommitLogModel commitLogModel = new CommitLogModel();
        commitLogModel.setFileName(LogFileNameUtil.buildFirstCommitLogName());
        commitLogModel.setOffset(new AtomicInteger(0));
        commitLogModel.setOffsetLimit(BrokerConstants.COMMITLOG_DEFAULT_SIZE);
        myMqTopicModel.setCommitLog(commitLogModel);
        List<QueueModel> queueList = new ArrayList<>();
        for (int i = 0; i < createTopicReqDTO.getQueueSize(); i++) {
            QueueModel queueModel = new QueueModel();
            queueModel.setId(i);
            queueModel.setFileName(LogFileNameUtil.buildFirstConsumeQueueName());
            queueModel.setOffsetLimit(BrokerConstants.CONSUME_QUEUE_DEFAULT_SIZE);
            queueModel.setLatestOffset(new AtomicInteger(0));
            queueModel.setLastOffset(0);
            queueList.add(queueModel);
        }
        myMqTopicModel.setQueueList(queueList);
        CommonCache.getMyMqTopicModels().add(myMqTopicModel);
        CommonCache.getMyMqTopicModelsMap().put(createTopicReqDTO.getTopic(), myMqTopicModel);
    }

    public static void main(String[] args) throws IOException {
        CreateTopicReqDTO createTopicReqDTO = new CreateTopicReqDTO();
        createTopicReqDTO.setTopic("test_topic");
        createTopicReqDTO.setQueueSize(3);

        CommonCache.getGlobalProperties().setMyMqHome("F:\\maven-proj\\mq");
        createTopicFile(createTopicReqDTO);
    }
}
